--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit 2b975bd4e6380d73efc1eb33f8f27923fd5de982
Parents : 195395b
Author : Ivan <ivan@quad4.io>
Signature : Signature validation error
Date : 2026-05-03T00:14:55-05:00
feat(legacy_migrator): implement automatic upstream folder migration and add related tests
Changes
2 files changed, 197 insertions(+), 0 deletions(-)
Diff
diff --git a/meshchatx/src/backend/legacy_migrator.py b/meshchatx/src/backend/legacy_migrator.py
index 77cdf6d3..6247b5ec 100644
--- a/meshchatx/src/backend/legacy_migrator.py
+++ b/meshchatx/src/backend/legacy_migrator.py
@@ -1,5 +1,6 @@
# SPDX-License-Identifier: 0BSD
+import logging
import os
import shutil
@@ -7,6 +8,8 @@ import RNS
LEGACY_DIR = ".reticulum-meshchat"
CURRENT_DIR = ".reticulum-meshchatx"
+UPSTREAM_DIR = "reticulum-meshchat"
+UPSTREAM_X_DIR = "reticulum-meshchatx"
def _basename_norm(p: str) -> str:
@@ -27,6 +30,17 @@ def paired_new_from_legacy(legacy_storage_path: str) -> str | None:
return os.path.join(parent, CURRENT_DIR)
+def _is_meshchatx_storage_basename(base: str) -> bool:
+ return base in (CURRENT_DIR, UPSTREAM_X_DIR)
+
+
+def paired_upstream_plain_from_meshchatx(meshchatx_path: str) -> str | None:
+ if not _is_meshchatx_storage_basename(_basename_norm(meshchatx_path)):
+ return None
+ parent = os.path.dirname(os.path.normpath(meshchatx_path))
+ return os.path.join(parent, UPSTREAM_DIR)
+
+
def storage_has_meshchat_data(storage_dir: str) -> bool:
if not storage_dir or not os.path.isdir(storage_dir):
return False
@@ -57,7 +71,40 @@ def resolve_startup_storage(request_dir: str) -> tuple[str, dict]:
if skip:
return planned, empty_ctx
+ skip_upstream_auto = os.environ.get(
+ "MESHCHAT_SKIP_UPSTREAM_FOLDER_MIGRATION",
+ "",
+ ).strip().lower() in ("1", "true", "yes")
+
base = _basename_norm(planned)
+ if (
+ not skip_upstream_auto
+ and _is_meshchatx_storage_basename(base)
+ and not storage_has_meshchat_data(planned)
+ ):
+ upstream_plain = paired_upstream_plain_from_meshchatx(planned)
+ if upstream_plain and storage_has_meshchat_data(upstream_plain):
+ try:
+ migrate_legacy_to_target(upstream_plain, planned)
+ logging.getLogger(__name__).info(
+ "Auto-copied upstream storage %s -> %s",
+ upstream_plain,
+ planned,
+ )
+ return planned, {
+ **empty_ctx,
+ "did_auto_upstream_folder_copy": True,
+ "upstream_copy_source": upstream_plain,
+ "upstream_copy_target": planned,
+ }
+ except OSError:
+ logging.getLogger(__name__).warning(
+ "Upstream folder auto-migration failed (%s -> %s)",
+ upstream_plain,
+ planned,
+ exc_info=True,
+ )
+
if base == CURRENT_DIR and not storage_has_meshchat_data(planned):
legacy = paired_legacy_from_new(planned)
if legacy and storage_has_meshchat_data(legacy):
diff --git a/tests/backend/test_legacy_migrator.py b/tests/backend/test_legacy_migrator.py
index c43d376d..3db40d27 100644
--- a/tests/backend/test_legacy_migrator.py
+++ b/tests/backend/test_legacy_migrator.py
@@ -6,6 +6,7 @@ import sqlite3
import stat
import tempfile
from pathlib import Path
+from unittest.mock import patch
import pytest
import RNS
@@ -15,12 +16,15 @@ from hypothesis import strategies as st
from meshchatx.src.backend.legacy_migrator import (
CURRENT_DIR,
LEGACY_DIR,
+ UPSTREAM_DIR,
+ UPSTREAM_X_DIR,
assert_migration_context_paths,
copy_legacy_storage_tree,
fresh_storage_at_target,
migrate_legacy_to_target,
paired_legacy_from_new,
paired_new_from_legacy,
+ paired_upstream_plain_from_meshchatx,
resolve_startup_storage,
storage_has_meshchat_data,
)
@@ -453,3 +457,149 @@ def test_resolve_identities_only_no_root_identity(monkeypatch):
assert ctx["show_choice"] is True
finally:
shutil.rmtree(root, ignore_errors=True)
+
+
+def test_paired_upstream_plain_from_meshchatx_paths():
+ p_dot = os.path.join("/tmp", "o", CURRENT_DIR)
+ assert paired_upstream_plain_from_meshchatx(p_dot) == os.path.join(
+ "/tmp", "o", UPSTREAM_DIR
+ )
+ p_plain = os.path.join("/tmp", "o", UPSTREAM_X_DIR)
+ assert paired_upstream_plain_from_meshchatx(p_plain) == os.path.join(
+ "/tmp", "o", UPSTREAM_DIR
+ )
+ assert paired_upstream_plain_from_meshchatx("/tmp/storage") is None
+
+
+def test_auto_upstream_plain_folder_to_dot_meshchatx(monkeypatch):
+ root = tempfile.mkdtemp()
+ try:
+ new_home = os.path.join(root, CURRENT_DIR)
+ plain = os.path.join(root, UPSTREAM_DIR)
+ os.makedirs(plain, exist_ok=True)
+ with open(os.path.join(plain, "identity"), "wb") as f:
+ f.write(RNS.Identity(create_keys=True).get_private_key())
+ os.makedirs(new_home, exist_ok=True)
+ monkeypatch.delenv("MESHCHAT_SKIP_LEGACY_MIGRATION_UI", raising=False)
+ monkeypatch.delenv("MESHCHAT_SKIP_UPSTREAM_FOLDER_MIGRATION", raising=False)
+ eff, ctx = resolve_startup_storage(new_home)
+ assert eff == os.path.abspath(new_home)
+ assert ctx.get("did_auto_upstream_folder_copy") is True
+ assert ctx.get("show_choice") is not True
+ assert storage_has_meshchat_data(new_home)
+ assert os.path.isfile(os.path.join(new_home, "identity"))
+ finally:
+ shutil.rmtree(root, ignore_errors=True)
+
+
+def test_auto_upstream_plain_folder_to_plain_meshchatx(monkeypatch):
+ root = tempfile.mkdtemp()
+ try:
+ new_home = os.path.join(root, UPSTREAM_X_DIR)
+ plain = os.path.join(root, UPSTREAM_DIR)
+ os.makedirs(plain, exist_ok=True)
+ with open(os.path.join(plain, "identity"), "wb") as f:
+ f.write(RNS.Identity(create_keys=True).get_private_key())
+ os.makedirs(new_home, exist_ok=True)
+ monkeypatch.delenv("MESHCHAT_SKIP_LEGACY_MIGRATION_UI", raising=False)
+ monkeypatch.delenv("MESHCHAT_SKIP_UPSTREAM_FOLDER_MIGRATION", raising=False)
+ eff, ctx = resolve_startup_storage(new_home)
+ assert eff == os.path.abspath(new_home)
+ assert ctx.get("did_auto_upstream_folder_copy") is True
+ assert storage_has_meshchat_data(new_home)
+ finally:
+ shutil.rmtree(root, ignore_errors=True)
+
+
+def test_auto_upstream_skipped_by_env(monkeypatch):
+ root = tempfile.mkdtemp()
+ try:
+ new_home = os.path.join(root, CURRENT_DIR)
+ plain = os.path.join(root, UPSTREAM_DIR)
+ os.makedirs(plain, exist_ok=True)
+ with open(os.path.join(plain, "identity"), "wb") as f:
+ f.write(RNS.Identity(create_keys=True).get_private_key())
+ os.makedirs(new_home, exist_ok=True)
+ leg = os.path.join(root, LEGACY_DIR)
+ os.makedirs(leg, exist_ok=True)
+ with open(os.path.join(leg, "identity"), "wb") as f:
+ f.write(RNS.Identity(create_keys=True).get_private_key())
+ monkeypatch.delenv("MESHCHAT_SKIP_LEGACY_MIGRATION_UI", raising=False)
+ monkeypatch.setenv("MESHCHAT_SKIP_UPSTREAM_FOLDER_MIGRATION", "1")
+ eff, ctx = resolve_startup_storage(new_home)
+ assert eff == leg
+ assert ctx.get("did_auto_upstream_folder_copy") is not True
+ assert ctx["show_choice"] is True
+ finally:
+ shutil.rmtree(root, ignore_errors=True)
+
+
+def test_auto_upstream_skipped_when_target_already_has_data(monkeypatch):
+ root = tempfile.mkdtemp()
+ try:
+ new_home = os.path.join(root, CURRENT_DIR)
+ plain = os.path.join(root, UPSTREAM_DIR)
+ os.makedirs(plain, exist_ok=True)
+ with open(os.path.join(plain, "identity"), "wb") as f:
+ f.write(RNS.Identity(create_keys=True).get_private_key())
+ os.makedirs(new_home, exist_ok=True)
+ with open(os.path.join(new_home, "identity"), "wb") as f:
+ f.write(RNS.Identity(create_keys=True).get_private_key())
+ monkeypatch.delenv("MESHCHAT_SKIP_LEGACY_MIGRATION_UI", raising=False)
+ monkeypatch.delenv("MESHCHAT_SKIP_UPSTREAM_FOLDER_MIGRATION", raising=False)
+ eff, ctx = resolve_startup_storage(new_home)
+ assert eff == os.path.abspath(new_home)
+ assert ctx.get("did_auto_upstream_folder_copy") is not True
+ assert ctx.get("show_choice") is not True
+ finally:
+ shutil.rmtree(root, ignore_errors=True)
+
+
+def test_auto_upstream_prefers_plain_over_dot_legacy_redirect(monkeypatch):
+ root = tempfile.mkdtemp()
+ try:
+ new_home = os.path.join(root, CURRENT_DIR)
+ plain = os.path.join(root, UPSTREAM_DIR)
+ os.makedirs(plain, exist_ok=True)
+ with open(os.path.join(plain, "identity"), "wb") as f:
+ f.write(RNS.Identity(create_keys=True).get_private_key())
+ leg = os.path.join(root, LEGACY_DIR)
+ os.makedirs(leg, exist_ok=True)
+ with open(os.path.join(leg, "identity"), "wb") as f:
+ f.write(RNS.Identity(create_keys=True).get_private_key())
+ os.makedirs(new_home, exist_ok=True)
+ monkeypatch.delenv("MESHCHAT_SKIP_LEGACY_MIGRATION_UI", raising=False)
+ monkeypatch.delenv("MESHCHAT_SKIP_UPSTREAM_FOLDER_MIGRATION", raising=False)
+ eff, ctx = resolve_startup_storage(new_home)
+ assert eff == os.path.abspath(new_home)
+ assert ctx.get("did_auto_upstream_folder_copy") is True
+ assert ctx.get("show_choice") is not True
+ finally:
+ shutil.rmtree(root, ignore_errors=True)
+
+
+def test_auto_upstream_oserror_falls_through_to_dot_legacy(monkeypatch):
+ root = tempfile.mkdtemp()
+ try:
+ new_home = os.path.join(root, CURRENT_DIR)
+ plain = os.path.join(root, UPSTREAM_DIR)
+ os.makedirs(plain, exist_ok=True)
+ with open(os.path.join(plain, "identity"), "wb") as f:
+ f.write(RNS.Identity(create_keys=True).get_private_key())
+ leg = os.path.join(root, LEGACY_DIR)
+ os.makedirs(leg, exist_ok=True)
+ with open(os.path.join(leg, "identity"), "wb") as f:
+ f.write(RNS.Identity(create_keys=True).get_private_key())
+ os.makedirs(new_home, exist_ok=True)
+ monkeypatch.delenv("MESHCHAT_SKIP_LEGACY_MIGRATION_UI", raising=False)
+ monkeypatch.delenv("MESHCHAT_SKIP_UPSTREAM_FOLDER_MIGRATION", raising=False)
+ with patch(
+ "meshchatx.src.backend.legacy_migrator.migrate_legacy_to_target",
+ side_effect=OSError("simulated copy failure"),
+ ):
+ eff, ctx = resolve_startup_storage(new_home)
+ assert eff == leg
+ assert ctx["show_choice"] is True
+ assert ctx.get("did_auto_upstream_folder_copy") is not True
+ finally:
+ shutil.rmtree(root, ignore_errors=True)
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────